package midas.mw.crt;

import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;

import midas.mw.cfg.CfgError;
import midas.mw.cfg.CfgLogging;
import midas.mw.cfg.log.Logger;
import midas.mw.crt.MD.NodeListMD;
import midas.mw.crt.MD.PriorityTable;
import midas.mw.crt.MD.TimeQueue;
import midas.mw.crt.interfaces.CrtCallback;
import midas.mw.crt.common.IPMessage;

/**
 * This class manages the IPMessages that are sent between nodes in the network
 * and do the store and forward of these messages based in the specified parameters,
 * such as priority and time to live <p>     
 * MessageDispatcher maintains two data structures: one for all the messages sorted
 * by node identifier (NodeListMD) and other that stores the messages to manage
 * ttls and expired messages (TimeQueue). <p>   
 */
public class MessageDispatcher extends TimerTask 
{
	/**
	 * Structure that stores all the messages that are going to be sent.
	 * 
	 * All messages are stored by destination node identifier and each node 
	 * has a Priority Table to manage the sending order.   
	 */
	private NodeListMD nodes;
	
	/**
	 * Simple structure composed by a ArralyList that is in charge of managing 
	 * all the messages that are stored in the NodeListMD.
	 * It manages the time to live (ttl) of messages and delete them when they 
	 * expire.  
	 */
	private TimeQueue times;
	
	/**
	 * Instance of the Communication & Routing Component (CRT) running in the node.
	 */
	private CrtImpl crt;
	
	/**
	 * Period of time to check for expired messages expressed in milliseconds
	 */
	private static final int checkTiming = 10 * 1000;
	
	/**
	 * Unique IPMessage identifier
	 */
	private long messageId = 0;
	
	/**
	 * This hashtable connects the identifier of the IPMessage with the IPMessage itself.
	 */
	private Hashtable messageList;
	
	
	/**
	 * This hashtable stores the different threads started to wait a response to an IPMessage
	 */
	//private Hashtable waitingThreads;
	
	/**
	 * This hashtable stores the different threads started to wait a response to an IPMessage
	 */
	private Hashtable searchingThreads;
	
        /**
	 * Timer to check periodically the expired messages
	 */
        private Timer timer;
	
	
	/**
	 * Constructor - Create the Message Dispatcher and initializes local variables
	 * 
	 * @param crt Instance of the CRT running in the node.
	 */
	public MessageDispatcher(CrtImpl crt) 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher(CrtImpl)");
		
		this.crt 		= crt;
		nodes 			= new NodeListMD();
		messageList 	= new Hashtable();
		//waitingThreads 	= new Hashtable();
		searchingThreads 	= new Hashtable();
		times 			= new TimeQueue();
		timer 			= new Timer();
		timer.schedule(this, 0, checkTiming);
		
	}
		
	/**
	 * Thread associated the the Timer to check periodically the expired messages
	 */
	public void run()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.run()");
		
		//logger.debug(className + ".run()");
		//***********************************************************/
		if (!times.isEmpty())
		{
			/*Date utilDate = new Date(); //fecha actual
			long lnMilisegundos = utilDate.getTime();
			Time time = new Time(lnMilisegundos);*/
			//nodes.printStruct();
			List deletedmsgs = times.deleteExpiredMessages();
			if (!(deletedmsgs.isEmpty()))
				messagesExpired(deletedmsgs);
		}
	}
	
    /**
     * Stops the Timer, causing it to stop sending action events to its listeners.
     * It clear the hashtables associated to the object too
     *
     */
    public void stopMessageDispatcher()
    {
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.stopMessageDispatcher()");    	
    	
    	nodes.getNodes().clear();    	
		messageList.clear();		
		//waitingThreads.clear();
		searchingThreads.clear();		
		times.getTimes().clear();    	
		timer.purge();
    	timer.cancel();
    }
	
    
    /**
     * Stores the IPMessage in the NodeListMD and in the TimeQueue structures.
     * 
     * @param ipMessage IP message to be stored.
     */
	private long enqueueMessage(IPMessage ipMessage)
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.enqueueMessage(IPMessage)");
		
		//Esta funcion la ha podido llamar
		//   -desde el propio middleware
		//   -si se es un nodo proxy al recibir un mensaje de un nodo JXTA/JXME
		//Si se trata de un nodo normal que intenta mandar un mensaje a un nodo JXME tiene que cambiar 
		//el nodo destino por el del proxy e incluir el destino final autentico
		if (crt.getNodeIdentifierSolver().getNodeIdToIpConnectedPeers().containsKey(ipMessage.getDestinationNodeId())
				&& crt.getNodeIdentifierSolver().isJXMENode(ipMessage.getDestinationNodeId())
				&& ((crt.getConstants().getNodeType() != Constants.CENTRAL_NODE)
				&& (crt.getConstants().getNodeType() != Constants.GATEWAY_NODE)))
		{				
			ipMessage.setFinalDestinationNodeId(ipMessage.getDestinationNodeId());						
			ipMessage.setDestinationNodeId(crt.getJxtaPlatform().getCentralNodeId());		
		}		
		//Si se trata de un CN puede que se trate de un mensaje redirigido
		if (ipMessage.getTotalLength() == 0)
			ipMessage.setTotalLength(ipMessage.getMessageContent().length);	
		synchronized (nodes.getNodeQueue(ipMessage.getDestinationNodeId()))
		{
			//System.out.println("DESPUES DE SINCRONIZAR ENQUE");
			/*System.out.println("MEMORIA USADA ANTES DE ALMACENAR: " + (Runtime.getRuntime().totalMemory()-Runtime.getRuntime().freeMemory()));
			//System.out.println("MEMORIA LIBRE ANTES DE ALMACENAR: " + Runtime.getRuntime().freeMemory());*/
			//Si se trata de un mensaje que viene de un nodo JXTA/JXME y somos un CN ya tiene
			//su propio identificador			
			if (ipMessage.getMessageId() == null)
				ipMessage.setMessageId(getMessageId());			
			nodes.insertIPMessage(ipMessage);
			times.insertOrderMessage(ipMessage);
						
			//Si no es un mensaje reenviado en el caso de un CN,
			//o es un mensaje generado en un UN
			//if (ipMessage.getFinalDestinationNodeId() == null)
			if (ipMessage.getSourceNodeId().equals(crt.getOwnNodeId()))
				messageList.put(ipMessage.getMessageId(), ipMessage);
			/*System.out.println("MEMORIA USADA DESPUES DE ALMACENAR: " + (Runtime.getRuntime().totalMemory()-Runtime.getRuntime().freeMemory()));
			//System.out.println("MEMORIA LIBRE DESPUES DE ALMACENAR: " + Runtime.getRuntime().freeMemory());*/
			//System.out.println("FIN DE SINCRONIZAR ENQUEUE");
		}		
		return ipMessage.getMessageId().longValue();
	}	
	
	/**
	 * Provides the identifier associated to an IPMessage
	 * 
	 * @return IPMessage identifier
	 */
	private Long getMessageId()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getMessageId()");
		
		return new Long(messageId++);
	}
	
		/**
	 * Sends a IPMessage 
	 * 
	 * @param ipMessage The IPMessage to be sent
	 */
	public long sendMessage(IPMessage ipMessage)
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.sendMessage(" + ipMessage.msgToString() + ")");
							
		long messageId = enqueueMessage(ipMessage);
		startSendingMessages(ipMessage.getDestinationNodeId());
		return messageId;		
	}
	
	/**
	 * Deletes the expired messages. 
	 * When a message in the NodeListMD has expired the CRT will communicate it
	 * to the source application.
	 *   
	 * @param deletedMessages List of expired messages
	 */
	private void messagesExpired (List deletedMessages)
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.messagesExpired(List)");		
		
		Iterator iter = deletedMessages.iterator();
		while(iter.hasNext())
		{
			IPMessage m = (IPMessage) iter.next();
			//Last chance to send the message
			synchronized (nodes.getNodeQueue(m.getDestinationNodeId()))
			{
				/*try {
					crt.getPacketForwarder().sendMessage(m);
				} catch (CrtException crte) {}*/
				String ident = m.getSourceUserId(); 
				nodes.deleteMessage(m);
				if (m.getFinalDestinationNodeId() == null)					
					messageList.remove(m.getMessageId());
				
				//waitingThreads.remove(m.getMessageId());											    			
				SearchThread st = (SearchThread) searchingThreads.get(m.getDestinationNodeId());
				if (st != null)
					st.stopThread();
				
				stopWaiting(m.getDestinationNodeId());				
				
				//First, we look for the destination among the registered components (but not the MW)
				CrtCallback crtUser = null;
				crtUser = (CrtCallback) crt.getCRTUser(ident);
				if (crtUser != null)
					crtUser.processUndeliveredMessage(m);
				else
				{
					//We look for the destination between components
					try {
						crtUser = (CrtCallback) crt.getMiddlewareInterface().getRunningComponent(ident);
					} catch (ClassCastException cce){}
					if (crtUser != null)
						crtUser.processUndeliveredMessage(m);
					//We look for the destination between MIDAS applications					
					else							
						CfgError.errorReporting(CfgError.ERR_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessageDispatcher.messagesExpired(List). " + 
								"The source application is not registered", null);					
				}							
			}
		}
	}
	
	/**
	 * Gets the structure of messages sorted by node identifier
	 * 
	 * @return The structure of messages sorted by node identifier
	 */
	public NodeListMD getNodeList()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getNodeList()");
		
		return nodes;
	}
	
	/**
	 * Gets the structure that manages the ttls of stored messages
	 * 
	 * @return The structure that manages the ttls of stored messages
	 */
	public TimeQueue getTimeQueue()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getTimeQueue()");
		
		return times;
	}
	
	/**
	 * Gets the list of messages based on their identifiers
	 * 
	 * @return The list of messages based on their identifiers
	 */
	public Hashtable getMessageList	()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getMessageList()");
		
		return messageList;
	}
	
	/**
	 * Gets the list of threads waiting for responses
	 * 
	 * @return The list of threads waiting for responses
	 */
	/*public Hashtable getWaitingThreads()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getWaitingThreads()");
		
		return waitingThreads;
	}*/
	
	/**
	 * Gets the list of threads waiting for searches
	 * 
	 * @return The list of threads waiting for searches
	 */
	public Hashtable getSearchingThreads()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.getSearchingThreads()");
		
		return searchingThreads;
	}
	
/**
	 * This function starts the thread to send enqueued messages to the node with the specified ID
	 * @param nodeId Node ID of the destination node
	 */
	public void startSendingMessages(String nodeId)
	{
    	CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.startSendingMessages(" + nodeId + ")");
    	
		Object obj = nodes.getNodeQueue(nodeId);
		if (obj instanceof PriorityTable)
		{
			PriorityTable pt = (PriorityTable)obj;			
			synchronized (pt)
			{				
				if (!pt.isEmptyTable())
				{
					QueueThread qt = pt.getQueueThread();
					if (qt == null || !qt.isAlive())
						qt = pt.setQueueThread(crt, nodeId);					
					if (!qt.isAlive())
						qt.start();				
				}				
			}
		}
	}
	
	public void stopWaiting(String nodeId)
	{
		Object obj = nodes.getNodeQueue(nodeId);
		if (obj instanceof PriorityTable)
		{
			PriorityTable pt = (PriorityTable)obj;
			//PriorityTable pt = (PriorityTable)nodes.getNodeQueue(nodeId);
			//System.out.println("ANTES DE SINCRONIZAR EL EL ARRANQUE DEL THREAD");
			synchronized (pt)
			{
				//System.out.println("DESPUES DE SINCRONIZAR EL EL ARRANQUE DEL THREAD");
				if (!pt.isEmptyTable())
				{
					QueueThread qt = pt.getQueueThread();
					if (qt != null)
						qt.stopWaiting();		
				}
			}
		}
	}
}
